package com.xf.day05

import org.apache.spark.{SparkConf, SparkContext}

object TestFold2 {
  def main(args: Array[String]): Unit = {
    // 创建 SparkConf 对象
    val conf = new SparkConf()
      .setAppName("WordCount")
      .setMaster("local[4]")
      .set("spark.ui.port", "8080")
      .set("spark.driver.host", "127.0.0.1")

    // 创建 SparkContext 对象
    val sc = new SparkContext(conf)

    val RddInt = sc.parallelize(List(1, 2, 3, 4, 5, 6, 2, 5, 1))

    // 查看实际分区数
    println(s"实际分区数: ${RddInt.getNumPartitions}")

    println("=== 分区数据分布 ===")
    RddInt.mapPartitionsWithIndex { (partitionIndex, iterator) =>
      val partitionData = iterator.toList
      println(s"分区 $partitionIndex: ${partitionData.mkString("[", ", ", "]")}")
      partitionData.iterator
    }.count()

    println("\n=== Fold 操作执行过程 ===")

    // 自定义 fold 操作来显示详细的聚合过程
    val foldInt = RddInt.mapPartitionsWithIndex { (partitionIndex, iterator) =>
      // 分区内的初始值
      val zeroValue = 2
      var partitionResult = zeroValue
      val elements = iterator.toList

      println(s"\n--- 分区 $partitionIndex 开始聚合 ---")
      println(s"初始值: $partitionResult")
      println(s"分区数据: ${elements.mkString("[", ", ", "]")}")

      // 分区内聚合
      for (element <- elements) {
        val oldValue = partitionResult
        partitionResult = oldValue + element
        println(s"  $oldValue + $element = $partitionResult")
      }

      println(s"分区 $partitionIndex 最终结果: $partitionResult")
      Iterator(partitionResult)
    }.collect()

    // 分区间聚合
    println(s"\n--- 分区间聚合 ---")
    var globalResult = 2  // 全局初始值
    println(s"全局初始值: $globalResult")

    for ((partitionResult, i) <- foldInt.zipWithIndex) {
      val oldValue = globalResult
      globalResult = oldValue + partitionResult
      println(s"  $oldValue + 分区$i 结果($partitionResult) = $globalResult")
    }

    println(s"\n最终结果: $globalResult")

    // 验证：使用原始的 fold 方法
    val originalFoldResult = RddInt.fold(4)((x, y) => x + y)
    println(s"原始 fold 方法结果: $originalFoldResult")

    sc.stop()
  }

}
